/*************************************************************************
 Copyright (C) Unpublished Unis Software, Inc. All rights reserved.
 Unis Software, Inc., Confidential and Proprietary.

 This software is subject to copyright protection
 under the laws of the Public of China and other countries.

 Unless otherwise explicitly stated, this software is provided
 by Unis "AS IS".
 *************************************************************************/
package com.unis.service.socketio;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.text.MessageFormat;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.Cache;
import org.springframework.cache.Cache.ValueWrapper;
import org.springframework.cache.CacheManager;

import com.corundumstudio.socketio.AckRequest;
import com.corundumstudio.socketio.Configuration;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.listener.ConnectListener;
import com.corundumstudio.socketio.listener.DataListener;
import com.corundumstudio.socketio.listener.DisconnectListener;
import com.unis.model.account.SysUser;

/**
 * @author CSJ (email:raulcsj@163.com)
 * @since 2013-1-6
 */
public class SocketIOService {
    private static final Logger LOGGER = LoggerFactory.getLogger(SocketIOService.class);

    private int port = 9092;

    @Value("${encache.name}")
    private String cacheName;
    @Value("${new.Message}")
    private String newMessage;

    @Resource
    private CacheManager cacheManager;

    /****************************************
     * Event 常量
     ****************************************/
    private static final String EVT_JOIN = "join";

    public SocketIOService() {}

    public SocketIOService(int port) {
        this.port = port;
    }

    private void cahceClient(SysUser sysuser, SocketIOClient client) {
        Cache cache = cacheManager.getCache(cacheName);
        // cache.put(sysuser.getUsername(), client);
        cache.put(sysuser.getUserId(), client);
        LOGGER.info("用户['" + sysuser.getUsername() + "']" + "进入消息队列!");
    }

    /****************************************
     * 启动SocketIO Server
     * 
     * @throws Exception
     ****************************************/
    public void init() throws Exception {
        final ExecutorService exec = Executors.newFixedThreadPool(1);
        exec.execute(new Runnable() {
            @Override
            public void run() {
                Configuration config = new Configuration();
                config.setPort(port);

                final SocketIOServer server = new SocketIOServer(config);

                server.addConnectListener(new ConnectListener() {
                    @Override
                    public void onConnect(SocketIOClient client) {
                        SocketAddress address = client.getRemoteAddress();
                        InetSocketAddress inetSocketAddress = (InetSocketAddress) address;
                        LOGGER.info("Weclome guest from " + inetSocketAddress.getHostName() + " : "
                                + inetSocketAddress.getPort());
                    }
                });
                server.addDisconnectListener(new DisconnectListener() {
                    @Override
                    public void onDisconnect(SocketIOClient client) {
                        SocketAddress address = client.getRemoteAddress();
                        InetSocketAddress inetSocketAddress = (InetSocketAddress) address;
                        LOGGER.info("Byebye guest from " + inetSocketAddress.getHostName() + " : "
                                + inetSocketAddress.getPort());
                    }
                });
                server.addMessageListener(new DataListener<String>() {
                    @Override
                    public void onData(SocketIOClient client, String msgs, AckRequest ackRequest) {
                        Cache cache = cacheManager.getCache(cacheName);
                        String[] msgArray = msgs.split(",");
                        SocketIOClient destClient = null;
                        for (String msg : msgArray) {
                            ValueWrapper valueWrapper = cache.get(msg.trim());
                            if (valueWrapper != null) {
                                destClient = (SocketIOClient) valueWrapper.get();
                            }
                            if (destClient != null) {
                                destClient.sendMessage(MessageFormat.format(newMessage, ""));
                            }
                        }
                    }
                });
                /**************************************************************
                 * 自定义事件消息处理(Customer event handler define)
                 *************************************************************/
                // 统计加入人员
                server.addEventListener(EVT_JOIN, SysUser.class, new DataListener<SysUser>() {
                    @Override
                    public void onData(SocketIOClient client, SysUser user, AckRequest ackRequest) {
                        cahceClient(user, client);
                    }
                });

                try {
                    server.start();
                    TimeUnit.DAYS.sleep(Long.MAX_VALUE);
                } catch (Exception e) {
                    LOGGER.error("Socket.iO Server 启动失败: " + e.getMessage());
                } finally {
                    server.stop();
                    exec.shutdown();
                }
            }
        });
    }

}
